package com.lcache.core.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lcache.common.api.ApiResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.buffer.ByteBuf;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.ClientException;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: lcache
 * @description: key处理类
 * @author: Alex Wu
 * @createDate: 2025-04-23 11:35
 **/

@Slf4j
public class HotKeyHandler  extends SimpleChannelInboundHandler<WebSocketFrame>  {

    private static final BlockingQueue<String> hotKeyQueue = new LinkedBlockingQueue<>();
    private static final ConcurrentMap<String, CacheValue> hotKeyResultMap = new ConcurrentHashMap<>();

    private final List<String> allServers;
    private final Map<String, Channel> serverChannels;

    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();


    private final AtomicBoolean loggedIn = new AtomicBoolean(false);

    public HotKeyHandler(List<String> allServers, Map<String, Channel> serverChannels) {
        this.allServers = allServers;
        this.serverChannels = serverChannels;
    }

    /**
     * TCP 建立连接成功，未初始化完成
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame) throws Exception {
        if (webSocketFrame instanceof TextWebSocketFrame) {
            String text = ((TextWebSocketFrame) webSocketFrame).text();
            log.info("收到消息: {}" , text);

            JSONObject json = JSONObject.parseObject(text);
            String type = json.getString("type");
            if ("login".equals(type)) {
                handleLogin(channelHandlerContext, json);
            } else if ("hotkey".equals(type)) {
                handleHotKeyMessage(channelHandlerContext, json);
            }
        } else if (webSocketFrame instanceof PongWebSocketFrame) {
            log.debug("收到心跳 Pong");
        }


    }

    private void handleHotKeyMessage(ChannelHandlerContext ctx, JSONObject json) throws ClientException {
        String keys = json.getString("keys");
        log.info("收到hotkey: {}", keys);
        if(StringUtils.isNotBlank(keys)){
            // 1. 缓存到本地，并设置过期时间 5 分钟后失效
            long expireAt = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5);
            String[] keysArr = keys.split(",");
            for (String rawKey : keysArr) {
                 String key = rawKey.trim();
                if (key.isEmpty()) continue;

                    // 2. 如果本地缓存中没有，就放进去一个空值占位，表示已见过
                hotKeyResultMap.computeIfAbsent(key, k -> new CacheValue(null, expireAt));
                log.debug("Key {} 插入本地缓存，expireAt={}", key, expireAt);

            }
        }


    }

    private void handleLogin(ChannelHandlerContext ctx, JSONObject json) {
        String success = json.getString("success");
        log.info("收到登录响应: {}", success);
    }


    private void startLogin(ChannelHandlerContext ctx) {
        if(loggedIn.compareAndSet(false, true)){
            String uid = "u" + ThreadLocalRandom.current().nextInt(100000, 999999);
            JSONObject msg = new JSONObject();
            msg.put("type", "login");
            msg.put("userId", uid);
            msg.put("username", uid);
            msg.put("password", "password123");

            ChannelFuture future = ctx.channel().writeAndFlush(new TextWebSocketFrame(msg.toJSONString()));
            log.info(future.channel().id().asLongText() + " " + future.isSuccess() + "");
            log.info("已发送登录请求: " + msg);
        }
    }

    private void startDispatcher() {
        scheduler.scheduleAtFixedRate(() -> {
            try {
                List<String> batchKeys = new ArrayList<>();
                hotKeyQueue.drainTo(batchKeys, 100);

                if (batchKeys.isEmpty()) return;

                Map<String, List<String>> serverKeyMap = new HashMap<>();

                for (String key : batchKeys) {
                    int idx = Math.abs(key.hashCode()) % allServers.size();
                    String targetServer = allServers.get(idx);
                    serverKeyMap.computeIfAbsent(targetServer, k -> new ArrayList<>()).add(key);
                }

                for (Map.Entry<String, List<String>> entry : serverKeyMap.entrySet()) {
                    String server = entry.getKey();
                    List<String> keys = entry.getValue();
                    Channel channel = serverChannels.get(server);
                    if (channel != null && channel.isActive()) {
                        sendHotKeyCheck(channel, keys);
                    }
                }

            } catch (Exception e) {
                log.error("分发hotKey失败", e);
            }
        }, 0, 500, TimeUnit.MILLISECONDS);
    }

    public void sendHotKeyCheck(Channel channel,  List<String> keys) {
        JSONObject hotKeyMsg = new JSONObject();
        hotKeyMsg.put("type", "hotkey");
        hotKeyMsg.put("keys", String.join(",", keys));
        channel.writeAndFlush(new TextWebSocketFrame(hotKeyMsg.toJSONString()));
        log.info("已发送热点检测: " + hotKeyMsg);
    }


    private void startHeartbeat(ChannelHandlerContext ctx) {
        scheduler.scheduleAtFixedRate(() ->
                ctx.writeAndFlush(new PingWebSocketFrame()), 30, 30, TimeUnit.SECONDS);
    }

    public static void submitHotKey(String key) {
        hotKeyQueue.offer(key);
    }

    public static Object getHotKeyResult(String key) {
        CacheValue cv = hotKeyResultMap.get(key);
        if (cv == null || cv.isExpired()) {
            hotKeyResultMap.remove(key);
            return null;
        }
        return cv.value;
    }

    public static boolean isHotKey(String key) {
        return getHotKeyResult(key)!=null? true:false;
    }

    /**
     * 握手完成
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) {
            startLogin(ctx);
            startDispatcher();
            startHeartbeat(ctx);
        }else{
            super.userEventTriggered(ctx, evt);
        }
    }


    // 新增：清理过期并返回存活 keys
    public static Set<String> cleanupAndGetLiveKeys() {
        long now = System.currentTimeMillis();
        Set<String> live = new HashSet<>();
        for (Iterator<Map.Entry<String, CacheValue>> it = hotKeyResultMap.entrySet().iterator(); it.hasNext(); ) {
            Map.Entry<String, CacheValue> e = it.next();
            if (e.getValue().isExpired()) {
                it.remove();
            } else {
                live.add(e.getKey());
            }
        }
        return live;
    }

    static class CacheValue {
        Object value;
        long expireAt;

        public CacheValue(Object value, long expireAt) {
            this.value = value;
            this.expireAt = expireAt;
        }

        boolean isExpired() {
            return System.currentTimeMillis() > expireAt;
        }
    }
}
